package zunge.twidel.typo;

import java.io.BufferedWriter;
import java.io.FileWriter;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;

import zunge.tool.db.sql.DbConn;

public class Main_Labeling {
	
	static final String labelFile = "/Users/bliu1/Desktop/labels.txt";
	
	static String farHost = "brutus.isr.cs.cmu.edu";
	static String farDb = "oneweek";
	static String farUser = "bliu1";
	static String farPswd = "87Clear#";
	
	
	static String tableName = "typorepost_new";//"typorepost";
	static TypoCheck tc;
	
	static int K = 3; //search distance
	
	static final int T = 3; //thread count
	
	public static void main(String args[]){
		if(args.length>0){
			farUser = args[0];
			farPswd = args[1];
			tableName = args[2];
			K = Integer.parseInt(args[3]);
		}
		tc = new TypoCheck(5, 0.4, 5, 4);
		
		try{
			List<Long> userList = getUsers();
			
			
			List<UserThread> threads = new ArrayList<UserThread>(T);
			
			//threads.add(new UserThread(-1, userList.subList(0, 500)));
			
			
			int step = userList.size() / (T-1);
			if(T>1){
				for(int i = 0;i < T-1;++i){
					threads.add(new UserThread(i, userList.subList(i*step, (i+1)*step)));
				}
				threads.add(new UserThread(T-1, userList.subList((T-1)*step, userList.size())));
			}
			else{
				threads.add(new UserThread(0,userList));
			}
			
			//CountdownThread ct = new CountdownThread();ct.start();
			for(int i = 0;i < T;++i){
				threads.get(i).start();
			}
			
		}catch(Exception e){e.printStackTrace();}
	}
	
	static List<Long> getUsers() throws Exception{
		HashSet<Long> users = new HashSet<Long>(150000);
		DbConn farConn = new DbConn(farHost, farDb, farUser, farPswd);
		
		String query = "select distinct user from dttweet";
		ResultSet rs = farConn.executeQuery(query);
		while(rs.next()){
			users.add(rs.getLong("user"));
		}
		rs.close();
		
		query = "SELECT dttweet.user FROM dttweet INNER JOIN "+
			tableName+" ON dttweet.tweetID = "+tableName+".tweetId";
		rs = farConn.executeQuery(query);
		while(rs.next()){
			users.remove(rs.getLong("user"));
		}
		rs.close();
		
		farConn.close();
		
		List<Long> userList = new ArrayList<Long>(users);
		System.out.println("UserCount: "+userList.size());
		//Thread.sleep(Integer.MAX_VALUE);
		
		return userList;
	}
	
	/**
	 * Process a subgroup of user
	 * @author bliu1
	 *
	 */
	public static class UserThread extends Thread{
		DbConn farConn;
		List<Long> userList;
		int label;
		
		UserThread(int label, List<Long> users){
			this.label = label;
			this.userList = users;
		}



		@Override
		public void run() {
			if(userList==null || userList.size()==0){return;}
			
			try{
				farConn = new DbConn(farHost, farDb, farUser, farPswd);
				
				PreparedStatement ps = farConn.conn.prepareStatement(
						"insert into "+tableName+" (tweetID, label, subTweetId) values (?,?,?)");
				for(int i = 0;i < userList.size();++i){
					long userId = userList.get(i);
					List<Tweet> userTweets = getUserTweetsSorted(userId);
					processUser(userTweets);
					for (Tweet tweet:userTweets){
						if(tweet.isDel){
							ps.setLong(1, tweet.tweetId);
							ps.setInt(2,tweet.label);
							ps.setLong(3,tweet.subsequentTweetId);
							farConn.executeUpdate(ps);
						}
					}
				}
				
				farConn.close();
				
			}catch(Exception e){e.printStackTrace();}
		}
			
		void processUser(List<Tweet> tweets){
			if(tweets == null || tweets.size()==0){return;}
			if(tweets.size()==1){ tweets.get(0).label=TypoCheck.TYPE_NORMAL;return;}
			
			for(int i = 0;i < tweets.size()-1;++i){
				Tweet prev = tweets.get(i);
				if(prev.isDel){
					for(int j = i+1;j < Math.min(i+K+1,tweets.size());++j){
						Tweet curr = tweets.get(j);
						int label = tc.typoCheck(curr.tweetText, prev.tweetText);
						if(label != TypoCheck.TYPE_NORMAL){
							prev.label = label;
							prev.subsequentTweetId = curr.tweetId;
						}
					}
				}
			}
		}
			
		List<Tweet> getUserTweetsSorted(long userId) throws SQLException, ParseException{
			List<Tweet> tweets = new ArrayList<Tweet>();
			
			getTweet("select tweetID,createdAt,text from dttweet where user = "+userId, true, tweets);
			getTweet("select tweetID,createdAt,text from undttweet where user = "+userId, false, tweets);
			
			Collections.sort(tweets);
			return tweets;
		}
		
		SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.0");
		void getTweet(String query, boolean isd, List<Tweet> tweets) throws SQLException, ParseException{
			ResultSet rs = farConn.executeQuery(query);
			while(rs.next()){
				try{
					tweets.add(new Tweet(rs.getLong("tweetID"), 
						sdf.parse(rs.getString("createdAt")),
						rs.getString("text"), 
						isd));
				}
				catch(ParseException e){
					System.err.println("TweetID="+rs.getLong("tweetID"));
					e.printStackTrace();
				}
			}
			rs.close();
		}

	}
	
	public static class CountdownThread extends Thread{
		DbConn farConn;
		Calendar startTime;
		
		CountdownThread(){
			startTime = Calendar.getInstance();
		}
		
		
		static final int total = 1617672;
		@Override
		public void run() {
			try{
				farConn = new DbConn(farHost, farDb, farUser, farPswd);
				while(true)
				{	
					String query = "select count(*) from typorepost";
					ResultSet rs = farConn.executeQuery(query);
					rs.next();
					int cur = rs.getInt("count(*)")+1;
					rs.close();
					
					if(cur==total){farConn.close();return;}
					Calendar curTime = Calendar.getInstance();
					long dist = curTime.getTimeInMillis() - startTime.getTimeInMillis();
					long esti = dist *(total-cur) / cur;
					System.out.println("TimeLeft: "+(int)(esti/(60*3600))+" minutes");
					Thread.sleep(30*1000);
				}
				
			}catch(Exception e){e.printStackTrace();}
		}
	}
}


